package day03;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * @Description: keyedProcessFunction例子
 * @Author: ZYX
 * @Date: 2022/2/10 11:16
 * @Version: 1.0
 */
public class Demo04 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.socketTextStream("localhost",9999)
                .keyBy(obj -> 1)
                .process(new MyKeyed())
                .print();

        env.execute();
    }

    public static class MyKeyed extends KeyedProcessFunction<Integer,String,String> {
        @Override
        public void processElement(String value, Context context, Collector<String> out) throws Exception {
            // 当前机器时间
            long time = context.timerService().currentProcessingTime();
            out.collect("元素"+value+"在"+new Timestamp(time) +"时间到达");

            // 注册一个10秒之后的定时器
            long timeStat = time + 10 * 1000L;
            context.timerService().registerProcessingTimeTimer(timeStat);


        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            out.collect("定时器触发了！触发时间："+new Timestamp(timestamp));
        }
    }

}
